package com.xl.flinkdemo.beans;

import org.apache.flink.streaming.api.functions.source.SourceFunction;

import java.util.HashMap;
import java.util.Random;

/**
 * @ClassName MySensor
 * @Description 自定义数据源类
 * @Author hxl
 * @Date 2021/4/14 16:04
 * Version 1.0
 **/
public class MySensor implements SourceFunction<SensorReding> {

  //定义个标识位，用来控制数据的产生

  private boolean running = true;

  @Override
  public void run(SourceContext<SensorReding> ctx) throws Exception {

    Random random = new Random();
    //设置10个传感器的初始温度

    HashMap<String,Double> sensorTempMap = new HashMap<String, Double>();
    for (int i=0;i<10;i++){
      sensorTempMap.put("sensor_"+i+1,random.nextGaussian() * 10);
    }

    while (running) {
      for (String sensorId:sensorTempMap.keySet()){
        //在当前温度上随机波动
        Double newtemp = sensorTempMap.get(sensorId) + random.nextGaussian();
        sensorTempMap.put(sensorId,newtemp);
        ctx.collect(new SensorReding(sensorId,System.currentTimeMillis(),newtemp));
      }
      //控制输出频率
        Thread.sleep(1000L);
      }

  }

  @Override
  public void cancel() {
    running = false;
  }

}
